package com.pisolution.scheduler.runtime.worker.processor;

import com.pisolution.scheduler.core.tuple.Tuple3;
import com.pisolution.scheduler.core.utils.JsonUtils;
import com.pisolution.scheduler.core.utils.OSUtils;
import com.pisolution.scheduler.datasource.instance.ExecutionState;
import com.pisolution.scheduler.remote.command.Command;
import com.pisolution.scheduler.remote.command.CommandType;
import com.pisolution.scheduler.remote.processor.NettyRequestProcessor;
import com.pisolution.scheduler.remote.utils.Host;
import com.pisolution.scheduler.runtime.SchedulerContext;
import com.pisolution.scheduler.runtime.context.TaskExecuteCommand;
import com.pisolution.scheduler.runtime.context.TaskExecutionContext;
import com.pisolution.scheduler.runtime.worker.WorkerConfig;
import com.pisolution.scheduler.runtime.worker.cache.ResponseCache;
import com.pisolution.scheduler.runtime.worker.cache.TaskExecutionContextCache;
import com.pisolution.scheduler.runtime.worker.cache.TaskExecutionContextCacheService;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 处理管理节点移交给工作节点的任务执行请求 请求处理逻辑详见具体的处理方法描述.
 *
 * @see #process(Channel, Command)
 * @author sora
 * @version 0.1.0
 */
public class TaskExecuteProcessor implements NettyRequestProcessor {
    private static final Logger LOGGER = LoggerFactory.getLogger(TaskExecuteProcessor.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于读取工作节点的配置信息. */
    private final WorkerConfig config;

    /** 用于缓存任务执行期的上下文环境. */
    private final TaskExecutionContextCache cache;

    /** 用于缓存接收指令的信道信息. */
    private final TaskCallbackService service;

    /** 用于缓存工作节点反馈给管理节点的指令. */
    private final ResponseCache responseCache;

    /** 用于创建任务执行环境. */
    private final ExecutorService executor;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化 {@link TaskExecuteProcessor}.
     */
    public TaskExecuteProcessor() {
        this.config = SchedulerContext.get(WorkerConfig.class);
        this.cache = SchedulerContext.get(TaskExecutionContextCacheService.class);
        this.service = SchedulerContext.get(TaskCallbackService.class);
        this.responseCache = SchedulerContext.get(ResponseCache.class);
        this.executor = Executors.newFixedThreadPool(
            this.config.getWorkerExecutorThreads(),
            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("worker-executor-thread").build()
        );
    }

    // -------------------------------------------------------------------------------------------------
    //  请求处理
    // -------------------------------------------------------------------------------------------------

    @Override
    public void process(Channel channel, Command command) {
        // 1. 校验接收到的指令类型是否与待处理的指令类型一致.
        Preconditions.checkArgument(
            CommandType.TASK_EXECUTE_REQUEST == command.getType(),
            "Received unrecognized command with type %s",
            command.getType()
        );

        // 2. 反序列化指令内容.
        final TaskExecuteCommand cmd = JsonUtils.deserialize(command.getBody(), TaskExecuteCommand.class);

        LOGGER.info("Received the interaction command {}.", cmd);

        // 3. 如果无法反序列化出指令内容 则忽略该指令并予以提示.
        if (cmd == null) {
            LOGGER.error("The task execute request command is null.");

            return;
        }

        // 4. 解析出任务执行的上下文环境.
        final TaskExecutionContext context = JsonUtils.deserialize(cmd.getContext(), TaskExecutionContext.class);

        // 5. 如果上下文环境无法成功解析 则予以提示并忽略该指令.
        if (context == null) {
            LOGGER.error("The task execution context is null.");

            return;
        }

        // 6. 缓存待执行任务的上下文环境.
        final TaskExecutionContext cachedContext = new TaskExecutionContext();

        cachedContext.setId(context.getId());
        this.cache.cache(cachedContext);

        // 7. 更新上下文环境中的主机网络地址等信息.
        context.setHost(OSUtils.getHost() + ":" + this.config.getPort());
        context.setStartTime(new Date());
        context.setLogPath("/home" + String.join(
            "/",
            String.valueOf(context.getUserId()),
            "scheduler",
            String.valueOf(context.getProjectId()),
            String.valueOf(context.getWorkflowId()),
            String.valueOf(context.getWorkflowInstanceId()),
            String.valueOf(context.getId()),
            "task.log"
        ));

        // 8. 缓存接收该指令的信道以便使用它传递反馈信息.
        this.service.addChannel(context.getId(), new Tuple3<>(command.getOpaque(), Host.of(channel), channel));

        // 9. 构建确认指令以反馈给管理节点 工作节点已经收到任务执行请求并进行了执行前的预处理.
        final TaskAckCommand ackCommand = new TaskAckCommand();

        ackCommand.setId(context.getId());
        ackCommand.setHost(context.getHost());
        ackCommand.setExecutePath(context.getExecutePath());
        ackCommand.setLogPath(context.getLogPath());
        ackCommand.setStatus(ExecutionState.RUNNING.getCode());
        ackCommand.setStartTime(context.getStartTime());

        this.responseCache.cacheAck(context.getId(), ackCommand.convert());
        this.service.sendAck(context.getId(), ackCommand.convert());

        // 10. 创建一个新线程以执行任务.
        this.executor.submit(new TaskExecuteThread(context, this.service));
    }
}
